package top.doe.flink.clickhouse;

import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import ru.ivi.opensource.flinkclickhousesink.ClickHouseSink;
import ru.ivi.opensource.flinkclickhousesink.model.ClickHouseClusterSettings;
import ru.ivi.opensource.flinkclickhousesink.model.ClickHouseSinkConst;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

public class Demo39_ClickHouse {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        Map<String, String> globalParameters = new HashMap<>();

        // ClickHouse集群属性
        globalParameters.put(ClickHouseClusterSettings.CLICKHOUSE_HOSTS, "doitedu01:8123");
        globalParameters.put(ClickHouseClusterSettings.CLICKHOUSE_USER, "hunter");
        globalParameters.put(ClickHouseClusterSettings.CLICKHOUSE_PASSWORD, "123456");

        // sink通用参数
        globalParameters.put(ClickHouseSinkConst.TIMEOUT_SEC, "30");
        globalParameters.put(ClickHouseSinkConst.FAILED_RECORDS_PATH, "/tmp/ck_fail/");
        globalParameters.put(ClickHouseSinkConst.NUM_WRITERS, "1");
        globalParameters.put(ClickHouseSinkConst.NUM_RETRIES, "3");
        globalParameters.put(ClickHouseSinkConst.QUEUE_MAX_CAPACITY, "10");
        globalParameters.put(ClickHouseSinkConst.IGNORING_CLICKHOUSE_SENDING_EXCEPTION_ENABLED, "true");

        // 设置job全局参数到env
        ParameterTool parameterTool = ParameterTool.fromMap(globalParameters);
        env.getConfig().setGlobalJobParameters(parameterTool);


        // 库表等配置
        Properties props = new Properties();
        props.put(ClickHouseSinkConst.TARGET_TABLE_NAME, "default.xx");
        props.put(ClickHouseSinkConst.MAX_BUFFER_SIZE, "1000");


        DataStreamSource<DataBean> ee = env.fromElements(new DataBean(6, "FF"));
        ee.addSink(new ClickHouseSink<>(props,new DataBeanConverter())).name("my-ck-sink");

        env.execute();


    }
}
